package simpledb.index;

import java.io.*;
import java.util.*;

import simpledb.common.Database;
import simpledb.common.Permissions;
import simpledb.execution.IndexPredicate;
import simpledb.execution.Predicate;
import simpledb.execution.Predicate.Op;
import simpledb.common.DbException;
import simpledb.common.Debug;
import simpledb.storage.*;
import simpledb.transaction.TransactionAbortedException;
import simpledb.transaction.TransactionId;

import javax.xml.crypto.dom.DOMURIReference;

/**
 * BTreeFile is an implementation of a DbFile that stores a B+ tree.
 * Specifically, it stores a pointer to a root page,
 * a set of internal pages, and a set of leaf pages, which contain a collection of tuples
 * in sorted order. BTreeFile works closely with BTreeLeafPage, BTreeInternalPage,
 * and BTreeRootPtrPage. The format of these pages is described in their constructors.
 * 
 * @see BTreeLeafPage#BTreeLeafPage
 * @see BTreeInternalPage#BTreeInternalPage
 * @see BTreeHeaderPage#BTreeHeaderPage
 * @see BTreeRootPtrPage#BTreeRootPtrPage
 * @author Becca Taft
 */
public class BTreeFile implements DbFile {

	private final File f;
	private final TupleDesc td;
	private final int tableid ;
	private final int keyField;

	/**
	 * Constructs a B+ tree file backed by the specified file.
	 * 
	 * @param f - the file that stores the on-disk backing store for this B+ tree
	 *            file.
	 * @param key - the field which index is keyed on
	 * @param td - the tuple descriptor of tuples in the file
	 */
	public BTreeFile(File f, int key, TupleDesc td) {
		this.f = f;
		this.tableid = f.getAbsoluteFile().hashCode();
		this.keyField = key;
		this.td = td;
	}

	/**
	 * Returns the File backing this BTreeFile on disk.
	 */
	public File getFile() {
		return f;
	}

	/**
	 * Returns an ID uniquely identifying this BTreeFile. Implementation note:
	 * you will need to generate this tableid somewhere and ensure that each
	 * BTreeFile has a "unique id," and that you always return the same value for
	 * a particular BTreeFile. We suggest hashing the absolute file name of the
	 * file underlying the BTreeFile, i.e. f.getAbsoluteFile().hashCode().
	 * 
	 * @return an ID uniquely identifying this BTreeFile.
	 */
	public int getId() {
		return tableid;
	}

	/**
	 * Returns the TupleDesc of the table stored in this DbFile.
	 * 
	 * @return TupleDesc of this DbFile.
	 */
	public TupleDesc getTupleDesc() {
		return td;
	}

	/**
	 * Read a page from the file on disk. This should not be called directly
	 * but should be called from the BufferPool via getPage()
	 * 
	 * @param pid - the id of the page to read from disk
	 * @return the page constructed from the contents on disk
	 */
	public Page readPage(PageId pid) {
		BTreePageId id = (BTreePageId) pid;

        try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(f))) {
            if (id.pgcateg() == BTreePageId.ROOT_PTR) {
                byte[] pageBuf = new byte[BTreeRootPtrPage.getPageSize()];
                int retval = bis.read(pageBuf, 0, BTreeRootPtrPage.getPageSize());
                if (retval == -1) {
                    throw new IllegalArgumentException("Read past end of table");
                }
                if (retval < BTreeRootPtrPage.getPageSize()) {
                    throw new IllegalArgumentException("Unable to read "
                            + BTreeRootPtrPage.getPageSize() + " bytes from BTreeFile");
                }
                Debug.log(1, "BTreeFile.readPage: read page %d", id.getPageNumber());
                return new BTreeRootPtrPage(id, pageBuf);
            } else {
                byte[] pageBuf = new byte[BufferPool.getPageSize()];
                if (bis.skip(BTreeRootPtrPage.getPageSize() + (long) (id.getPageNumber() - 1) * BufferPool.getPageSize()) !=
                        BTreeRootPtrPage.getPageSize() + (long) (id.getPageNumber() - 1) * BufferPool.getPageSize()) {
                    throw new IllegalArgumentException(
                            "Unable to seek to correct place in BTreeFile");
                }
                int retval = bis.read(pageBuf, 0, BufferPool.getPageSize());
                if (retval == -1) {
                    throw new IllegalArgumentException("Read past end of table");
                }
                if (retval < BufferPool.getPageSize()) {
                    throw new IllegalArgumentException("Unable to read "
                            + BufferPool.getPageSize() + " bytes from BTreeFile");
                }
                Debug.log(1, "BTreeFile.readPage: read page %d", id.getPageNumber());
                if (id.pgcateg() == BTreePageId.INTERNAL) {
                    return new BTreeInternalPage(id, pageBuf, keyField);
                } else if (id.pgcateg() == BTreePageId.LEAF) {
                    return new BTreeLeafPage(id, pageBuf, keyField);
                } else { // id.pgcateg() == BTreePageId.HEADER
                    return new BTreeHeaderPage(id, pageBuf);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // Close the file on success or error
        // Ignore failures closing the file
    }

	/**
	 * Write a page to disk.  This should not be called directly but should 
	 * be called from the BufferPool when pages are flushed to disk
	 * 
	 * @param page - the page to write to disk
	 */
	public void writePage(Page page) throws IOException {
		BTreePageId id = (BTreePageId) page.getId();
		
		byte[] data = page.getPageData();
		RandomAccessFile rf = new RandomAccessFile(f, "rw");
		if(id.pgcateg() == BTreePageId.ROOT_PTR) {
			rf.write(data);
			rf.close();
		}
		else {
			rf.seek(BTreeRootPtrPage.getPageSize() + (long) (page.getId().getPageNumber() - 1) * BufferPool.getPageSize());
			rf.write(data);
			rf.close();
		}
	}
	
	/**
	 * Returns the number of pages in this BTreeFile.
	 */
	public int numPages() {
		// we only ever write full pages
		return (int) ((f.length() - BTreeRootPtrPage.getPageSize())/ BufferPool.getPageSize());
	}

	/**
	 * Returns the index of the field that this B+ tree is keyed on
	 */
	public int keyField() {
		return keyField;
	}

	/**
	 * Recursive function which finds and locks the leaf page in the B+ tree corresponding to
	 * the left-most page possibly containing the key field f. It locks all internal
	 * nodes along the path to the leaf node with READ_ONLY permission, and locks the 
	 * leaf node with permission perm.
	 * 
	 * If f is null, it finds the left-most leaf page -- used for the iterator
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param pid - the current page being searched
	 * @param perm - the permissions with which to lock the leaf page
	 * @param f - the field to search for
	 * @return the left-most leaf page possibly containing the key field f
	 * 
	 */
	private BTreeLeafPage findLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid, Permissions perm,
                                       Field f)
					throws DbException, TransactionAbortedException {
		// System.out.println(tid);
		// some code goes here
		int type = pid.pgcateg();
		if (type==BTreePageId.LEAF) {
			return (BTreeLeafPage) getPage(tid, dirtypages, pid, perm);
		}
		BTreeInternalPage page = (BTreeInternalPage) getPage(tid, dirtypages, pid, Permissions.READ_ONLY);
		Iterator<BTreeEntry> iterator = page.iterator();
		BTreeEntry entry = null;
		while (iterator.hasNext()) {
			entry = iterator.next();
			// 如果为空，当前节点是遍历到的第一个节点，既最左节点，直接找左孩子
			if (f==null) {
				// 提取释放读锁
				// Database.getBufferPool().unsafeReleasePage(tid, page.pid);
				return findLeafPage(tid, dirtypages, entry.getLeftChild(), perm, f);
			}
			// 当前节点的值比大于等于要找的，肯定往左边找
			if (entry.getKey().compare(Op.GREATER_THAN_OR_EQ, f)) {
				// Database.getBufferPool().unsafeReleasePage(tid, page.pid);
				return findLeafPage(tid, dirtypages, entry.getLeftChild(), perm, f);
			}
		}
		assert entry != null;
		// 遍历完了还没找到比当前f大的，往最右节点找
		return findLeafPage(tid, dirtypages, entry.getRightChild(), perm, f);
	}
	
	/**
	 * Convenience method to find a leaf page when there is no dirtypages HashMap.
	 * Used by the BTreeFile iterator.
	 * @see #findLeafPage(TransactionId, Map, BTreePageId, Permissions, Field)
	 * 
	 * @param tid - the transaction id
	 * @param pid - the current page being searched
	 * @param f - the field to search for
	 * @return the left-most leaf page possibly containing the key field f
	 * 
	 */
	BTreeLeafPage findLeafPage(TransactionId tid, BTreePageId pid,
                               Field f)
					throws DbException, TransactionAbortedException {
		return findLeafPage(tid, new HashMap<>(), pid, Permissions.READ_ONLY, f);
	}

	/**
	 * Split a leaf page to make room for new tuples and recursively split the parent node
	 * as needed to accommodate a new entry. The new entry should have a key matching the key field
	 * of the first tuple in the right-hand page (the key is "copied up"), and child pointers 
	 * pointing to the two leaf pages resulting from the split.  Update sibling pointers and parent 
	 * pointers as needed.  
	 * 
	 * Return the leaf page into which a new tuple with key field "field" should be inserted.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the leaf page to split
	 * @param field - the key field of the tuple to be inserted after the split is complete. Necessary to know
	 * which of the two pages to return.
	 * @see #getParentWithEmptySlots(TransactionId, Map, BTreePageId, Field)
	 * 
	 * @return the leaf page into which the new tuple should be inserted
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public BTreeLeafPage splitLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeLeafPage page, Field field)
			throws DbException, IOException, TransactionAbortedException {
		// some code goes here
        //
        // Split the leaf page by adding a new page on the right of the existing
		// page and moving half of the tuples to the new page.  Copy the middle key up
		// into the parent page, and recursively split the parent as needed to accommodate
		// the new entry.  getParentWithEmtpySlots() will be useful here.  Don't forget to update
		// the sibling pointers of all the affected leaf pages.  Return the page into which a 
		// tuple with the given key field should be inserted.

		// 创建的新节点
		BTreeLeafPage rightPage = (BTreeLeafPage) getEmptyPage(tid, dirtypages, BTreePageId.LEAF);
		BTreePageId rightPageId = rightPage.getId();

		// 原节点最大元组数
		int maxTuples = page.getNumTuples();
		// 分裂后右页节点数
		int mid = (maxTuples+1) / 2;
		// 获取从右开始的迭代器
		Iterator<Tuple> iterator = page.reverseIterator();

		for (int i=0; i<mid; i++) {
			Tuple tuple = iterator.next();
			// 从左节点删除
			page.deleteTuple(tuple);
			// 放入右节点
			rightPage.insertTuple(tuple);
		}
		// 左节点id
		BTreePageId leftPageId = page.getId();
		BTreePageId rightSiblingId = page.getRightSiblingId();
		// System.out.println("原页右兄弟: " + (rightSiblingId != null ? rightSiblingId.getPageNumber() : "null"));
		System.out.println("[TID" + tid.getId() + "] 原页右兄弟: " + (rightSiblingId != null ? rightSiblingId.getPageNumber() : "null"));
		// 更新兄弟指针 - 原页右指针指向新页
		page.setRightSiblingId(rightPageId);
		System.out.println("[TID" + tid.getId() + "] 指针变更: 原页(" + leftPageId.getPageNumber() + ")右指针 -> 新页(" + rightPageId.getPageNumber() + ")");

		// 新页左指针指向原页
		rightPage.setLeftSiblingId(leftPageId);
		System.out.println("[TID" + tid.getId() + "] 指针变更: 新页(" + rightPageId.getPageNumber() + ")左指针 -> 原页(" + leftPageId.getPageNumber() + ")");

		// 新页右指针指向原右兄弟
		rightPage.setRightSiblingId(rightSiblingId);
		System.out.println("[TID" + tid.getId() + "] 指针变更: 新页(" + rightPageId.getPageNumber() + ")右指针 -> 原右兄弟(" + (rightSiblingId != null ? rightSiblingId.getPageNumber() : "null") + ")");

		// 页被修改了，放入脏页
		dirtypages.put(leftPageId, page);
		dirtypages.put(rightPageId, rightPage);

		// 更新原右节点的左孩子
		if (rightSiblingId!=null) {
			BTreeLeafPage rightSibling = (BTreeLeafPage) getPage(tid, dirtypages, rightSiblingId, Permissions.READ_WRITE);
			rightSibling.setLeftSiblingId(rightPageId);
			System.out.println("[TID" + tid.getId() + "] 指针变更: 原右兄弟(" + rightSiblingId.getPageNumber() + ")左指针 -> 新页(" + rightPageId.getPageNumber() + ")");
			dirtypages.put(rightSiblingId, rightSibling);
		}

		// 获取要提取上去的那个主键
		Field splitKey = rightPage.iterator().next().getField(page.keyField);
		// 获取第一个找到的空槽的parent
		BTreeInternalPage parent = getParentWithEmptySlots(tid, dirtypages, page.getParentId(), splitKey);
		// 生成新的entry
		BTreeEntry entry = new BTreeEntry(splitKey, leftPageId, rightPageId);
		parent.insertEntry(entry);
		// parent也被修改了，放入脏页
		dirtypages.put(parent.getId(), parent);

		// 更新两个节点的父指针,因为已经把entry插入到parent中，所有可以遍历修改
		updateParentPointers(tid, dirtypages, parent);

		// System.out.println ("分裂后当前节点条目数:" + page.getNumTuples() + ", 分裂后兄弟条目数:" + rightPage.getNumTuples());
		System.out.println("[TID" + tid.getId() + "] 分裂后节点ID-" + page.getId().getPageNumber() + " 条目数:" + page.getNumTuples() + ", 分裂后兄弟节点ID-" + rightPage.getId().getPageNumber() + " 条目数:" + rightPage.getNumTuples());


		// 如果要插入的键比当前的分裂节点小，插左边，否则插入右边
		if (field.compare(Op.LESS_THAN_OR_EQ, splitKey)) {
			return page;
		}else {
			return rightPage;
		}
	}
	
	/**
	 * Split an internal page to make room for new entries and recursively split its parent page
	 * as needed to accommodate a new entry. The new entry for the parent should have a key matching 
	 * the middle key in the original internal page being split (this key is "pushed up" to the parent). 
	 * The child pointers of the new parent entry should point to the two internal pages resulting 
	 * from the split. Update parent pointers as needed.
	 * 
	 * Return the internal page into which an entry with key field "field" should be inserted
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the internal page to split
	 * @param field - the key field of the entry to be inserted after the split is complete. Necessary to know
	 * which of the two pages to return.
	 * @see #getParentWithEmptySlots(TransactionId, Map, BTreePageId, Field)
	 * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
	 * 
	 * @return the internal page into which the new entry should be inserted
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public BTreeInternalPage splitInternalPage(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeInternalPage page, Field field) 
					throws DbException, IOException, TransactionAbortedException {
		// some code goes here
        //
        // Split the internal page by adding a new page on the right of the existing
		// page and moving half of the entries to the new page.  Push the middle key up
		// into the parent page, and recursively split the parent as needed to accommodate
		// the new entry.  getParentWithEmtpySlots() will be useful here.  Don't forget to update
		// the parent pointers of all the children moving to the new page.  updateParentPointers()
		// will be useful here.  Return the page into which an entry with the given key field
		// should be inserted.

		BTreeInternalPage rightPage = (BTreeInternalPage) getEmptyPage(tid, dirtypages, BTreePageId.INTERNAL);
		System.out.printf("[TID" + tid.getId() + "] 中间节点ID-%d 分裂开始，初始条目数为%d\n", page.getId().getPageNumber(), page.getNumEntries());
		Iterator<BTreeEntry> iterator = page.reverseIterator();
		int pos = 0;
		int target = (page.getNumEntries() + 1) / 2;
		// System.out.printf("中间节点ID-%d 目标移除元组数量%d\n", page.getId().getPageNumber(), target);
		for (; pos<target; pos++) {
			BTreeEntry entry = iterator.next();
			// 其实就是调用删除
			page.deleteKeyAndRightChild(entry);
			rightPage.insertEntry(entry);
		}
		// System.out.println(pos);
		// System.out.printf("中间节点ID-%d 分裂结束, 当前元组数量为%d\n", page.getId().getPageNumber(), page.getNumEntries());

		// 如果是比右侧最大的小于等于，那就插入左边，就让左边分裂一个给父节点
		BTreeEntry rentry = rightPage.iterator().next();
		BTreeEntry lentry = page.reverseIterator().next();

		BTreeInternalPage parent = null;

		if (field.compare(Op.LESS_THAN_OR_EQ, rentry.getKey())) { //应该插入左侧，那就让左侧分裂一个给parent
			page.deleteKeyAndRightChild(lentry);
			lentry.setLeftChild(page.getId());
			lentry.setRightChild(rightPage.getId());
			parent = getParentWithEmptySlots(tid, dirtypages, page.getParentId(), lentry.getKey());
			parent.insertEntry(lentry);
		}else { //应该插入右侧，让右侧分裂一个给parent
			rightPage.deleteKeyAndLeftChild(rentry);
			rentry.setLeftChild(page.getId());
			rentry.setRightChild(rightPage.getId());
			parent = getParentWithEmptySlots(tid, dirtypages, page.getParentId(), rentry.getKey());
			parent.insertEntry(rentry);
		}

		dirtypages.put(page.getId(), page);
		dirtypages.put(rightPage.getId(), rightPage);
		dirtypages.put(parent.getId(), parent);

		// 这三个页的孩子都变了，所以要更新，但是其实page页面只是变少了，他的孩子节点的parent一直是他，其实不用变
		// 注意，下边操作更新的是孩子到parent的，parent到孩子其实已经通过entry更新了,实测注释后不影响测试
		updateParentPointers(tid, dirtypages, parent);
		updateParentPointers(tid, dirtypages, page);
		updateParentPointers(tid, dirtypages, rightPage);

		// System.out.println("父节点更新后条目数:" + parent.getNumEntries() + ", 左侧子节点条目数:" + page.getNumEntries() + ", 右侧子节点条目数:" + rightPage.getNumEntries());
		System.out.println("[TID" + tid.getId() + "] 父节点ID-" + parent.getId().getPageNumber() + " 更新后条目数:" + parent.getNumEntries() + ", 左侧子节点ID-" + page.getId().getPageNumber() + " 条目数:" + page.getNumEntries() + ", 右侧子节点ID-" + rightPage.getId().getPageNumber() + " 条目数:" + rightPage.getNumEntries());

		if (field.compare(Op.LESS_THAN_OR_EQ, rentry.getKey())) {
			return page;
		}
		return rightPage;
	}
	
	/**
	 * Method to encapsulate the process of getting a parent page ready to accept new entries.
	 * This may mean creating a page to become the new root of the tree, splitting the existing 
	 * parent page if there are no empty slots, or simply locking and returning the existing parent page.
	 *
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param parentId - the id of the parent. May be an internal page or the RootPtr page
	 * @param field - the key of the entry which will be inserted. Needed in case the parent must be split
	 * to accommodate the new entry
	 * @return the parent page, guaranteed to have at least one empty slot
	 * @see #splitInternalPage(TransactionId, Map, BTreeInternalPage, Field)
	 *
	 * 返回一个有空闲槽位的父节点
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private BTreeInternalPage getParentWithEmptySlots(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreePageId parentId, Field field) throws DbException, IOException, TransactionAbortedException {
		
		BTreeInternalPage parent = null;
		
		// create a parent node if necessary
		// this will be the new root of the tree

		// 要分裂的节点的parent是BTreeRootPtrPage，代表着要分裂的节点是根节点，既B+树的root节点，因为是root了，只能创建新的root
		if(parentId.pgcateg() == BTreePageId.ROOT_PTR) {
			// 获取空页
			parent = (BTreeInternalPage) getEmptyPage(tid, dirtypages, BTreePageId.INTERNAL);

			// update the root pointer
			// 获取根信息页
			BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages,
					BTreeRootPtrPage.getId(tableid), Permissions.READ_WRITE);
			BTreePageId prevRootId = rootPtr.getRootId(); //save prev id before overwriting.
			// 更新根信息页中的根为新创建的页面
			rootPtr.setRootId(parent.getId());

			System.out.println("[TID" + tid.getId() + "] 新根节点ID-" + parent.getId().getPageNumber() + ", 元组数量: " + parent.getNumEntries());

			// update the previous root to now point to this new root.
			// 原来的根有了新的parent，就是新的根
			BTreePage prevRootPage = (BTreePage)getPage(tid, dirtypages, prevRootId, Permissions.READ_WRITE);
			prevRootPage.setParentId(parent.getId());
		}
		else { 
			// lock the parent page
			// 如果是普通节点，就获取parent对应的节点
			parent = (BTreeInternalPage) getPage(tid, dirtypages, parentId, 
					Permissions.READ_WRITE);
			System.out.println("[TID" + tid.getId() + "] 获取现有父节点, ID-" + parentId.getPageNumber() + ", 初始entry数量: " + parent.getNumEntries());
		}

		// split the parent if needed
		// 如果parent的空槽数量为0 需要分裂
		if(parent.getNumEmptySlots() == 0) {
			System.out.println("[TID" + tid.getId() + "] 父节点ID-" + parent.getId().getPageNumber() + " 空槽数量为0，需要分裂");
			parent = splitInternalPage(tid, dirtypages, parent, field);
		}
		// System.out.println("父节点无需分裂, 当前元组数量: " + parent.getNumEntries());
		System.out.println("[TID" + tid.getId() + "] 父节点ID-" + parent.getId().getPageNumber() + " 无需分裂, 当前元组数量: " + parent.getNumEntries());

		return parent;

	}

	/**
	 * Helper function to update the parent pointer of a node.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param pid - id of the parent node
	 * @param child - id of the child node to be updated with the parent pointer
	 * 更新child节点的父指针，保证双向指针
	 * @throws DbException
	 * @throws TransactionAbortedException
	 */
	private void updateParentPointer(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid, BTreePageId child)
			throws DbException, TransactionAbortedException {

		BTreePage p = (BTreePage) getPage(tid, dirtypages, child, Permissions.READ_ONLY);

		if(!p.getParentId().equals(pid)) {
			p = (BTreePage) getPage(tid, dirtypages, child, Permissions.READ_WRITE);
			p.setParentId(pid);
		}

	}
	
	/**
	 * Update the parent pointer of every child of the given page so that it correctly points to
	 * the parent
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the parent page
	 * @see #updateParentPointer(TransactionId, Map, BTreePageId, BTreePageId)
	 * 吧一个页面的所有的孩子节点指向当前页面本身,通过遍历BTreeEntry来实现
	 * @throws DbException
	 * @throws TransactionAbortedException
	 */
	private void updateParentPointers(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage page)
			throws DbException, TransactionAbortedException{
		Iterator<BTreeEntry> it = page.iterator();
		BTreePageId pid = page.getId();
		BTreeEntry e = null;
		while(it.hasNext()) {
			e = it.next();
			updateParentPointer(tid, dirtypages, pid, e.getLeftChild());
		}
		if(e != null) {
			updateParentPointer(tid, dirtypages, pid, e.getRightChild());
		}
	}
	
	/**
	 * Method to encapsulate the process of locking/fetching a page.  First the method checks the local 
	 * cache ("dirtypages"), and if it can't find the requested page there, it fetches it from the buffer pool.  
	 * It also adds pages to the dirtypages cache if they are fetched with read-write permission, since 
	 * presumably they will soon be dirtied by this transaction.
	 * 
	 * This method is needed to ensure that page updates are not lost if the same pages are
	 * accessed multiple times.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param pid - the id of the requested page
	 * @param perm - the requested permissions on the page
	 * @return the requested page
	 * 首先从脏页缓存中获取数据，确保事务能看到之前修改的数据，对于新页，如果是只读权限，不用放到脏页缓存中，因为只读不会修改数据，所以没必要
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	Page getPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid, Permissions perm)
			throws DbException, TransactionAbortedException {
		if(dirtypages.containsKey(pid)) {
			return dirtypages.get(pid);
		}
		else {
			Page p = null;
			p = Database.getBufferPool().getPage(tid, pid, perm);
			if(perm == Permissions.READ_WRITE) {
				dirtypages.put(pid, p);
			}
			return p;
		}
	}

	/**
	 * Insert a tuple into this BTreeFile, keeping the tuples in sorted order. 
	 * May cause pages to split if the page where tuple t belongs is full.
	 * 
	 * @param tid - the transaction id
	 * @param t - the tuple to insert
	 * @return a list of all pages that were dirtied by this operation. Could include
	 * many pages since parent pointers will need to be updated when an internal node splits.
	 * @see #splitLeafPage(TransactionId, Map, BTreeLeafPage, Field)
	 */
	public synchronized List<Page> insertTuple(TransactionId tid, Tuple t)
			throws DbException, IOException, TransactionAbortedException {
		Map<PageId, Page> dirtypages = new HashMap<>();

		// get a read lock on the root pointer page and use it to locate the root page
		// 当前的根页面是用读权限获取的
		BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
		BTreePageId rootId = rootPtr.getRootId();

		if(rootId == null) { // the root has just been created, so set the root pointer to point to it
			// 如果当前的根页面中没存root节点的信息，需要用创建新的root节点为叶子节点类型，同时把根页用读写权限读取，修改root节点指向
			rootId = new BTreePageId(tableid, numPages(), BTreePageId.LEAF);
			rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid), Permissions.READ_WRITE);
			rootPtr.setRootId(rootId);
		}

		// find and lock the left-most leaf page corresponding to the key field,
		// and split the leaf page if there are no more slots available
		// 选取最左边可以插入的页作为插入页
		BTreeLeafPage leafPage = findLeafPage(tid, dirtypages, rootId, Permissions.READ_WRITE, t.getField(keyField));
		System.out.println("[TID" + tid.getId() + "] [insertTuple] 找到叶子节点，ID-" + leafPage.getId().getPageNumber() + "，当前元组位数量: " + leafPage.getNumTuples());
		if(leafPage.getNumEmptySlots() == 0) {
			// 如果空间不足，分裂页
			leafPage = splitLeafPage(tid, dirtypages, leafPage, t.getField(keyField));	
		}

		// insert the tuple into the leaf page
		// 插入
		leafPage.insertTuple(t);
		// System.out.println("[insertTuple] 插入后节点条目数为: " + leafPage.getNumTuples());
		System.out.println("[TID" + tid.getId() + "] [insertTuple] 插入后节点ID-" + leafPage.getId().getPageNumber() + " 条目数为: " + leafPage.getNumTuples());
		return new ArrayList<>(dirtypages.values());
	}
	
	/**
	 * Handle the case when a B+ tree page becomes less than half full due to deletions.
	 * If one of its siblings has extra tuples/entries, redistribute those tuples/entries.
	 * Otherwise merge with one of the siblings. Update pointers as needed.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the page which is less than half full
	 * @see #handleMinOccupancyLeafPage(TransactionId, Map, BTreeLeafPage, BTreeInternalPage, BTreeEntry, BTreeEntry)
	 * @see #handleMinOccupancyInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeEntry, BTreeEntry)
	 * B + 树在删除操作后，处理页面填充率过低（低于一半）的核心逻辑，
	 * 通过重分配（redistribute） 或合并（merge） 操作维持 B 树的平衡性和查询效率。
	 * 若兄弟节点填充率超过一半，拉取一些来保持平衡
	 * 如果兄弟也少于一半，合并两个节点减少页面数
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private void handleMinOccupancyPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePage page)
			throws DbException, IOException, TransactionAbortedException {
		BTreePageId parentId = page.getParentId();
		BTreeEntry leftEntry = null;
		BTreeEntry rightEntry = null;
		BTreeInternalPage parent = null;

		// find the left and right siblings through the parent so we make sure they have
		// the same parent as the page. Find the entries in the parent corresponding to 
		// the page and siblings
		if(parentId.pgcateg() != BTreePageId.ROOT_PTR) {
			parent = (BTreeInternalPage) getPage(tid, dirtypages, parentId, Permissions.READ_WRITE);
			Iterator<BTreeEntry> ite = parent.iterator();
			// 遍历父节点中的所有条目，找到与当前页面相关的左右条目
			while(ite.hasNext()) {
				BTreeEntry e = ite.next();
				// 如果遍历到的父节点的某一个entry的左孩子是自己，说明找到右兄弟，找完了退出
				if(e.getLeftChild().equals(page.getId())) {
					rightEntry = e;
					break;
				}
				// 如果遍历到的entry的右孩子是自己，说明找到左兄弟，继续找右兄弟
				else if(e.getRightChild().equals(page.getId())) {
					leftEntry = e;
				}
			}
		}
		
		if(page.getId().pgcateg() == BTreePageId.LEAF) {
			handleMinOccupancyLeafPage(tid, dirtypages, (BTreeLeafPage) page, parent, leftEntry, rightEntry);
		}
		else { // BTreePageId.INTERNAL
			handleMinOccupancyInternalPage(tid, dirtypages, (BTreeInternalPage) page, parent, leftEntry, rightEntry);
		}
	}
	
	/**
	 * Handle the case when a leaf page becomes less than half full due to deletions.
	 * If one of its siblings has extra tuples, redistribute those tuples.
	 * Otherwise merge with one of the siblings. Update pointers as needed.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the leaf page which is less than half full
	 * @param parent - the parent of the leaf page
	 * @param leftEntry - the entry in the parent pointing to the given page and its left-sibling
	 * @param rightEntry - the entry in the parent pointing to the given page and its right-sibling
	 * @see #mergeLeafPages(TransactionId, Map, BTreeLeafPage, BTreeLeafPage, BTreeInternalPage, BTreeEntry)
	 * @see #stealFromLeafPage(BTreeLeafPage, BTreeLeafPage, BTreeInternalPage,  BTreeEntry, boolean)
	 * 处理叶子节点的平衡和空间问题
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private void handleMinOccupancyLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeLeafPage page,
			BTreeInternalPage parent, BTreeEntry leftEntry, BTreeEntry rightEntry) 
			throws DbException, IOException, TransactionAbortedException {
		BTreePageId leftSiblingId = null;
		BTreePageId rightSiblingId = null;
		// 若leftEntry为null，说明当前页面是最左侧的叶子节点（无左兄弟）；若rightEntry为null，说明是最右侧的叶子节点（无右兄弟）
		if(leftEntry != null) leftSiblingId = leftEntry.getLeftChild();
		if(rightEntry != null) rightSiblingId = rightEntry.getRightChild();

		// 最多允许一半空槽
		int maxEmptySlots = page.getMaxTuples() - page.getMaxTuples()/2; // ceiling
		// 先处理左兄弟， 左兄弟如果已经处理了，就不考虑右兄弟，已经两个平衡了，再来一个没法处理
		if(leftSiblingId != null) {
			BTreeLeafPage leftSibling = (BTreeLeafPage) getPage(tid, dirtypages, leftSiblingId, Permissions.READ_WRITE);
			// if the left sibling is at minimum occupancy, merge with it. Otherwise
			// steal some tuples from it
			// 左兄弟空槽位≥阈值 → 合并；否则 → 窃取元组
			if(leftSibling.getNumEmptySlots() >= maxEmptySlots) {
				mergeLeafPages(tid, dirtypages, leftSibling, page, parent, leftEntry);
			}
			else {
				stealFromLeafPage(page, leftSibling, parent, leftEntry, false);				
			}
		}
		else if(rightSiblingId != null) {	
			BTreeLeafPage rightSibling = (BTreeLeafPage) getPage(tid, dirtypages, rightSiblingId, Permissions.READ_WRITE);
			// if the right sibling is at minimum occupancy, merge with it. Otherwise
			// steal some tuples from it
			// 右兄弟空槽位≥阈值 → 合并；否则 → 窃取元组
			if(rightSibling.getNumEmptySlots() >= maxEmptySlots) {
				mergeLeafPages(tid, dirtypages, page, rightSibling, parent, rightEntry);
			}
			else {
				stealFromLeafPage(page, rightSibling, parent, rightEntry, true);				
			}
		}
	}
	
	/**
	 * Steal tuples from a sibling and copy them to the given page so that both pages are at least
	 * half full.  Update the parent's entry so that the key matches the key field of the first
	 * tuple in the right-hand page.
	 * 
	 * @param page - the leaf page which is less than half full
	 * @param sibling - the sibling which has tuples to spare
	 * @param parent - the parent of the two leaf pages
	 * @param entry - the entry in the parent pointing to the two leaf pages
	 * @param isRightSibling - whether the sibling is a right-sibling
	 *
	 * @throws DbException
	 */
	public void stealFromLeafPage(BTreeLeafPage page, BTreeLeafPage sibling,
			BTreeInternalPage parent, BTreeEntry entry, boolean isRightSibling) throws DbException {
		// some code goes here
        //
        // Move some of the tuples from the sibling to the page so
		// that the tuples are evenly distributed. Be sure to update
		// the corresponding parent entry.
		int pageNumTuples = page.getNumTuples();
		int totalTuples = pageNumTuples + sibling.getNumTuples();
		// 设置的右边多一个，所以如果当前节点是左节点，就少偷一个
		int target = isRightSibling ? (totalTuples)/2 : (totalTuples+1)/2;
		int moveCount = target - pageNumTuples;
		// 理论上不会出现，检查一下
		// if (moveCount<=0) {
		// 	return;
		// }
		// 若为左兄弟，要从反向迭代器获取
		Iterator<Tuple> iterator = isRightSibling? sibling.iterator() : sibling.reverseIterator();

		for (int i=0; i<moveCount; i++) {
			Tuple tuple = iterator.next();
			// 注意只能先删除后插入，因为插入操作会修改tuple的记录id，再删就报错了
			// 兄弟删除
			sibling.deleteTuple(tuple);
			// 当前节点添加
			page.insertTuple(tuple);
		}

		// 更新父节点条目：key为右页第一个元组的key
		BTreeLeafPage rightPage = isRightSibling ? sibling : page;

		Field newKey = rightPage.iterator().next().getField(rightPage.keyField);
		// 更新父节点的保存的索引
		entry.setKey(newKey);
		// 写到磁盘，注意，现在并没有修改entry的左孩子和右孩子，所以不需要更新指针，只是entry的key变了
		// 注意 在中间节点文件中，不是按照entry这种对象格式存的，用数组存的，所以需要调用updateEntry
		parent.updateEntry(entry);
		// System.out.println ("初始总条目数:" + totalTuples + ", 拉取后当前节点条目数:" + page.getNumTuples() + ", 拉取后兄弟条目数:" + sibling.getNumTuples());
		System.out.println("节点ID-" + page.getId().getPageNumber() + " 初始总条目数:" + totalTuples + ", 拉取后当前节点条目数:" + page.getNumTuples() + ", 拉取后兄弟节点ID-" + sibling.getId().getPageNumber() + " 条目数:" + sibling.getNumTuples());

	}

	/**
	 * Handle the case when an internal page becomes less than half full due to deletions.
	 * If one of its siblings has extra entries, redistribute those entries.
	 * Otherwise merge with one of the siblings. Update pointers as needed.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the internal page which is less than half full
	 * @param parent - the parent of the internal page
	 * @param leftEntry - the entry in the parent pointing to the given page and its left-sibling
	 * @param rightEntry - the entry in the parent pointing to the given page and its right-sibling
	 * @see #mergeInternalPages(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
	 * @see #stealFromLeftInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
	 * @see #stealFromRightInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
	 * 专门处理 B + 树中内部页面（非叶子节点，存储索引条目）因删除操作导致填充率低于一半的情况（即页面过于稀疏）
	 * 思路和叶子节点的方法一样，不做注释
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private void handleMinOccupancyInternalPage(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeInternalPage page, BTreeInternalPage parent, BTreeEntry leftEntry, BTreeEntry rightEntry) 
					throws DbException, IOException, TransactionAbortedException {
		// System.out.println("开始非叶节点处理");
		BTreePageId leftSiblingId = null;
		BTreePageId rightSiblingId = null;
		if(leftEntry != null) leftSiblingId = leftEntry.getLeftChild();
		if(rightEntry != null) rightSiblingId = rightEntry.getRightChild();
		
		int maxEmptySlots = page.getMaxEntries() - page.getMaxEntries()/2; // ceiling
		if(leftSiblingId != null) {
			BTreeInternalPage leftSibling = (BTreeInternalPage) getPage(tid, dirtypages, leftSiblingId, Permissions.READ_WRITE);
			// if the left sibling is at minimum occupancy, merge with it. Otherwise
			// steal some entries from it
			if(leftSibling.getNumEmptySlots() >= maxEmptySlots) {
				// 融合非页节点
				// System.out.println("融合非页节点");
				mergeInternalPages(tid, dirtypages, leftSibling, page, parent, leftEntry);
			}
			else {
				// System.out.println("平分非页节点");
				stealFromLeftInternalPage(tid, dirtypages, page, leftSibling, parent, leftEntry);
			}
		}
		else if(rightSiblingId != null) {
			BTreeInternalPage rightSibling = (BTreeInternalPage) getPage(tid, dirtypages, rightSiblingId, Permissions.READ_WRITE);
			// if the right sibling is at minimum occupancy, merge with it. Otherwise
			// steal some entries from it
			if(rightSibling.getNumEmptySlots() >= maxEmptySlots) {
				// System.out.println("融合非页节点");
				mergeInternalPages(tid, dirtypages, page, rightSibling, parent, rightEntry);
			}
			else {
				// System.out.println("平分非页节点");
				stealFromRightInternalPage(tid, dirtypages, page, rightSibling, parent, rightEntry);
			}
		}
	}
	
	/**
	 * Steal entries from the left sibling and copy them to the given page so that both pages are at least
	 * half full. Keys can be thought of as rotating through the parent entry, so the original key in the 
	 * parent is "pulled down" to the right-hand page, and the last key in the left-hand page is "pushed up"
	 * to the parent.  Update parent pointers as needed.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the internal page which is less than half full
	 * @param leftSibling - the left sibling which has entries to spare
	 * @param parent - the parent of the two internal pages
	 * @param parentEntry - the entry in the parent pointing to the two internal pages
	 * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
	 * 
	 * @throws DbException
	 * @throws TransactionAbortedException
	 */
	public void stealFromLeftInternalPage(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeInternalPage page, BTreeInternalPage leftSibling, BTreeInternalPage parent,
			BTreeEntry parentEntry) throws DbException, TransactionAbortedException {
		// some code goes here
        // Move some of the entries from the left sibling to the page so
		// that the entries are evenly distributed. Be sure to update
		// the corresponding parent entry. Be sure to update the parent
		// pointers of all children in the entries that were moved.
		int pageNumTuples = page.getNumEntries();
		int totalTuples = pageNumTuples + leftSibling.getNumEntries();
		// 因为是右边拉根，右边再放上去一个，所以不用刻意修改数量
		int target = 0;
		if ((totalTuples&1)==0) {
			target = totalTuples/2;
		}else {
			target = totalTuples/2;
		}
		int moveCount = target - pageNumTuples;
		// 理论上不会出现，检查一下
		// if (moveCount<=0) {
		// 	return;
		// }

		// 把父节点的key要放下来，因为还要换上去一个
		page.insertEntry(new BTreeEntry(parentEntry.getKey(), leftSibling.reverseIterator().next().getRightChild(), page.iterator().next().getLeftChild()));

		Iterator<BTreeEntry> iterator = leftSibling.reverseIterator();
		for (int i=0; i<moveCount; i++) {
			BTreeEntry entry = iterator.next();
			// 删除等于删除key和右指针，左指针不动
			leftSibling.deleteKeyAndRightChild(entry);
			// 当前节点插入
			page.insertEntry(entry);
		}

		BTreeEntry entry = page.iterator().next();
		// 右页的最左节点要换上去
		page.deleteKeyAndLeftChild(entry);
		parentEntry.setKey(entry.getKey());
		parent.updateEntry(parentEntry);

		dirtypages.put(page.getId(), page);
		dirtypages.put(leftSibling.getId(), leftSibling);
		dirtypages.put(parent.getId(), parent);

		// parent虽然修改了，但是孩子的parent指针没变，左兄弟只是少了，多的page才需要把新弄进来的parent指向自己
		updateParentPointers(tid, dirtypages, page);
		System.out.println ("节点ID：" + page.getId().getPageNumber() + "初始总条目数:" + totalTuples + ", 拉取后当前节点条目数:" + page.getNumEntries() + ", 拉取后左兄弟条目数:" + leftSibling.getNumEntries());
	}
	
	/**
	 * Steal entries from the right sibling and copy them to the given page so that both pages are at least
	 * half full. Keys can be thought of as rotating through the parent entry, so the original key in the 
	 * parent is "pulled down" to the left-hand page, and the last key in the right-hand page is "pushed up"
	 * to the parent.  Update parent pointers as needed.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param page - the internal page which is less than half full
	 * @param rightSibling - the right sibling which has entries to spare
	 * @param parent - the parent of the two internal pages
	 * @param parentEntry - the entry in the parent pointing to the two internal pages
	 * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
	 * 
	 * @throws DbException
	 * @throws TransactionAbortedException
	 */
	public void stealFromRightInternalPage(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeInternalPage page, BTreeInternalPage rightSibling, BTreeInternalPage parent,
			BTreeEntry parentEntry) throws DbException, TransactionAbortedException {
		// some code goes here
        // Move some of the entries from the right sibling to the page so
		// that the entries are evenly distributed. Be sure to update
		// the corresponding parent entry. Be sure to update the parent
		// pointers of all children in the entries that were moved.
		int pageNumTuples = page.getNumEntries();
		int totalTuples = pageNumTuples + rightSibling.getNumEntries();
		// 设置的右边多一个,所以不能多偷，因为右节点还要向上分出去一个
		int target = 0;
		if ((totalTuples&1)==1) {
			target = (totalTuples)/2;
		}else {
			target = (totalTuples)/2-1;
		}
		// 打印总条目数和计算出的目标条目数
		// System.out.println("总条目数: " + totalTuples + " (当前节点: " + pageNumTuples + ", 右兄弟节点: " + rightSibling.getNumEntries() + ")");
		// System.out.println("计算目标条目数: " + target + " (当前节点需要从右兄弟窃取 " + (target - pageNumTuples) + " 个条目)");
		// if (totalTuples)

		System.out.println("[TID" + tid.getId() + "] 节点ID-" + page.getId().getPageNumber() + " 初始条目数:" + page.getNumEntries() + ", 右兄弟节点ID-" + rightSibling.getId().getPageNumber() + " 初始条目数:" + rightSibling.getNumEntries() + "总条目数:" + totalTuples);
		System.out.println("[TID" + tid.getId() + "] 节点ID-" + page.getId().getPageNumber() + "计算目标条目数: " + target + " (当前节点需要从右兄弟窃取 " + (target - pageNumTuples) + " 个条目)");


		int moveCount = target - pageNumTuples;
		// 理论上不会出现，检查一下
		// 不能加这个，即使不从兄弟拉，也要像父节点拉取一个，兄弟节点再给父节点一个
		// if (moveCount<=0) {
		// 	return;
		// }

		// 把父节点的key要放下来，因为还要换上去一个
		page.insertEntry(new BTreeEntry(parentEntry.getKey(), page.reverseIterator().next().getRightChild(), rightSibling.iterator().next().getLeftChild()));

		Iterator<BTreeEntry> iterator = rightSibling.iterator();
		for (int i=0; i<moveCount; i++) {
			BTreeEntry entry = iterator.next();
			// 删除等于删除key和左指针，右指针不动
			rightSibling.deleteKeyAndLeftChild(entry);
			// 当前节点插入
			page.insertEntry(entry);
		}

		BTreeEntry entry = rightSibling.iterator().next();
		// 右页的最左节点要换上去
		rightSibling.deleteKeyAndLeftChild(entry);
		parentEntry.setKey(entry.getKey());
		parent.updateEntry(parentEntry);

		dirtypages.put(page.getId(), page);
		dirtypages.put(rightSibling.getId(), rightSibling);
		dirtypages.put(parent.getId(), parent);

		// parent虽然修改了，但是孩子的parent指针没变，左兄弟只是少了，多的page才需要把新弄进来的parent指向自己
		updateParentPointers(tid, dirtypages, page);
		updateParentPointers(tid, dirtypages, page);
		System.out.println ("[TID" + tid.getId() + "] 节点ID-" + page.getId().getPageNumber() + "初始总条目数:" + totalTuples + ", 拉取后当前节点条目数:" + page.getNumEntries() + ", 拉取后右兄弟" +"节点ID-" + rightSibling.getId().getPageNumber() + "条目数拉取后右兄弟条目数:" + rightSibling.getNumEntries());
	}
	
	/**
	 * Merge two leaf pages by moving all tuples from the right page to the left page. 
	 * Delete the corresponding key and right child pointer from the parent, and recursively 
	 * handle the case when the parent gets below minimum occupancy.
	 * Update sibling pointers as needed, and make the right page available for reuse.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param leftPage - the left leaf page
	 * @param rightPage - the right leaf page
	 * @param parent - the parent of the two pages
	 * @param parentEntry - the entry in the parent corresponding to the leftPage and rightPage
	 * @see #deleteParentEntry(TransactionId, Map, BTreePage, BTreeInternalPage, BTreeEntry)
	 * 
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public void mergeLeafPages(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeLeafPage leftPage, BTreeLeafPage rightPage, BTreeInternalPage parent, BTreeEntry parentEntry) 
					throws DbException, IOException, TransactionAbortedException {
		// System.out.println("触发融合操作");
		// some code goes here
        //
		// Move all the tuples from the right page to the left page, update
		// the sibling pointers, and make the right page available for reuse.
		// Delete the entry in the parent corresponding to the two pages that are merging -
		// deleteParentEntry() will be useful here
		// 移动右页所有元组到左页
		// System.out.println("触发页节点融合操作");
		// System.out.println ("开始融合叶节点，初始左兄弟元组数:" + leftPage.getNumTuples() + ", 右兄弟元组数:" + rightPage.getNumTuples() + ", 父亲节点条目数:" + parent.getNumEntries());
		System.out.println("[TID" + tid.getId() + "] 开始融合叶节点，左兄弟节点ID-" + leftPage.getId().getPageNumber() + " 初始元组数:" + leftPage.getNumTuples() + ", 右兄弟节点ID-" + rightPage.getId().getPageNumber() + " 初始元组数:" + rightPage.getNumTuples() + ", 父亲节点ID-" + parent.getId().getPageNumber() + " 初始条目数:" + parent.getNumEntries());



		Iterator<Tuple> iterator = rightPage.iterator();
		while (iterator.hasNext()) {
			Tuple t = iterator.next();
			rightPage.deleteTuple(t);
			leftPage.insertTuple(t);
		}
		BTreePageId rightSiblingId = rightPage.getRightSiblingId();
		// 左页的右兄弟设为右右兄弟
		leftPage.setRightSiblingId(rightSiblingId);
		System.out.println("[TID" + tid.getId() + "] 指针变更: 左页(" + leftPage.getId().getPageNumber() + ")右指针 -> 原右兄弟的右兄弟(" + (rightSiblingId != null ? rightSiblingId.getPageNumber() : "null") + ")");

		if (rightSiblingId!=null) {
			BTreeLeafPage rightSibling = (BTreeLeafPage) getPage(tid, dirtypages, rightSiblingId, Permissions.READ_WRITE);
			// 右右兄弟的左兄弟设为左页
			rightSibling.setLeftSiblingId(leftPage.getId());
			System.out.println("[TID" + tid.getId() + "] 指针变更: 右右兄弟(" + rightSiblingId.getPageNumber() + ")左指针 -> 左页(" + leftPage.getId().getPageNumber() + ")");
			dirtypages.put(rightSiblingId, rightSibling);
		}
		// 清空右页
		// System.out.println("清空右页，页号为："+ rightPage.getId().getPageNumber());
		setEmptyPage(tid, dirtypages, rightPage.getId().getPageNumber());

		// 删除对应的父节点条目和右指针，递归处理, 传入左页是防止删除的是根，让左页成为新的根
		// 因为右节点已经删除了，所以不需要更新指针啥的，且是叶子节点，虽然他们变了，但是左页没变，parent还是原来的那个，
		deleteParentEntry(tid, dirtypages, leftPage, parent, parentEntry);
		dirtypages.put(leftPage.getId(), leftPage);
		dirtypages.put(parent.getId(), parent);

		System.out.println("[TID" + tid.getId() + "] 叶节点融合后，左兄弟节点ID-" + leftPage.getId().getPageNumber() + " 元组数:" + leftPage.getNumTuples() + ", 父亲节点ID-" + parent.getId().getPageNumber() + " 条目数:" + parent.getNumEntries());
		// System.out.println ("叶节点融合后，左兄弟元组数:" + leftPage.getNumTuples() + ", 父亲节点条目数:" + parent.getNumEntries());
	}

	/**
	 * Merge two internal pages by moving all entries from the right page to the left page 
	 * and "pulling down" the corresponding key from the parent entry. 
	 * Delete the corresponding key and right child pointer from the parent, and recursively 
	 * handle the case when the parent gets below minimum occupancy.
	 * Update parent pointers as needed, and make the right page available for reuse.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param leftPage - the left internal page
	 * @param rightPage - the right internal page
	 * @param parent - the parent of the two pages
	 * @param parentEntry - the entry in the parent corresponding to the leftPage and rightPage
	 * @see #deleteParentEntry(TransactionId, Map, BTreePage, BTreeInternalPage, BTreeEntry)
	 * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
	 * 
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public void mergeInternalPages(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreeInternalPage leftPage, BTreeInternalPage rightPage, BTreeInternalPage parent, BTreeEntry parentEntry) 
					throws DbException, IOException, TransactionAbortedException {
		
		// some code goes here
        //
        // Move all the entries from the right page to the left page, update
		// the parent pointers of the children in the entries that were moved, 
		// and make the right page available for reuse
		// Delete the entry in the parent corresponding to the two pages that are merging -
		// deleteParentEntry() will be useful here
		// 拉取父节点的key到左页, 注意叶子节点不用拉取，但是要更改指针
		// System.out.println("触发中间节点融合操作");

		// System.out.println ("开始融合非叶节点，初始左兄弟条目数:" + leftPage.getNumEntries() + ", 右兄弟条目数:" + rightPage.getNumEntries() + ", 父亲节点条目数:" + parent.getNumEntries());
		System.out.println("[TID" + tid.getId() + "] 开始融合非叶节点，左兄弟节点ID-" + leftPage.getId().getPageNumber() + " 初始条目数:" + leftPage.getNumEntries() + ", 右兄弟节点ID-" + rightPage.getId().getPageNumber() + " 初始条目数:" + rightPage.getNumEntries() + ", 父亲节点ID-" + parent.getId().getPageNumber() + " 初始条目数:" + parent.getNumEntries());
		BTreeEntry pulledEntry = new BTreeEntry(parentEntry.getKey(), leftPage.reverseIterator().next().getRightChild(), rightPage.iterator().next().getLeftChild());
		leftPage.insertEntry(pulledEntry);

		// 移动右页所有条目到左页
		Iterator<BTreeEntry> iterator = rightPage.iterator();
		while (iterator.hasNext()) {
			BTreeEntry e = iterator.next();
			rightPage.deleteKeyAndLeftChild(e);
			leftPage.insertEntry(e);
		}

		// 更新移动条目的子节点的父指针，因为这些移动的都是代表一个个的页，所以需要更新
		updateParentPointers(tid, dirtypages, leftPage);
		// 释放右页
		setEmptyPage(tid, dirtypages, rightPage.getId().getPageNumber());

		// 删除父节点条目并处理可能的递归
		deleteParentEntry(tid, dirtypages, leftPage, parent, parentEntry);

		dirtypages.put(leftPage.getId(), leftPage);
		dirtypages.put(parent.getId(), parent);
		// System.out.println ("非叶节点融合后，左兄弟条目数:" + leftPage.getNumEntries() + ", 父亲节点条目数:" + parent.getNumEntries());
		System.out.println("[TID" + tid.getId() + "] 非叶节点融合后，左兄弟节点ID-" + leftPage.getId().getPageNumber() + " 条目数:" + leftPage.getNumEntries() + ", 父亲节点ID-" + parent.getId().getPageNumber() + " 条目数:" + parent.getNumEntries());
	}
	
	/**
	 * Method to encapsulate the process of deleting an entry (specifically the key and right child) 
	 * from a parent node.  If the parent becomes empty (no keys remaining), that indicates that it 
	 * was the root node and should be replaced by its one remaining child.  Otherwise, if it gets 
	 * below minimum occupancy for non-root internal nodes, it should steal from one of its siblings or 
	 * merge with a sibling.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param leftPage - the child remaining after the key and right child are deleted
	 * @param parent - the parent containing the entry to be deleted
	 * @param parentEntry - the entry to be deleted
	 * @see #handleMinOccupancyPage(TransactionId, Map, BTreePage)
	 * 封装从父节点中删除某个条目（具体指条目中的 “键” 和 “右子节点”）的过程。
	 * 若删除条目后，父节点变得为空（没有剩余的键）：说明该父节点是根节点，此时应将其剩余的一个子节点（leftPage）设为新的根节点。
	 * 若删除条目后，父节点是非根内部节点且填充率低于 “最小填充率”：则需要通过从兄弟节点窃取条目或与兄弟节点合并的方式，维持父节点的合理填充率。
	 * parentEntry：要从父节点中删除的条目（包含键和右子节点）
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private void deleteParentEntry(TransactionId tid, Map<PageId, Page> dirtypages,
			BTreePage leftPage, BTreeInternalPage parent, BTreeEntry parentEntry) 
					throws DbException, IOException, TransactionAbortedException {		
		
		// delete the entry in the parent.  If
		// the parent is below minimum occupancy, get some tuples from its siblings
		// or merge with one of the siblings
		// 删除右孩子
		parent.deleteKeyAndRightChild(parentEntry);
		System.out.printf("[TID" + tid.getId() + "] [deleteParentEntry]父节点ID-%d 删除节点后节点数量变为%d\n", parent.getId().getPageNumber(), parent.getNumEntries());
		int maxEmptySlots = parent.getMaxEntries() - parent.getMaxEntries()/2; // ceiling
		if(parent.getNumEmptySlots() == parent.getMaxEntries()) {
			// This was the last entry in the parent.
			// In this case, the parent (root node) should be deleted, and the merged 
			// page will become the new root
			// 父节点已无任何条目（空节点），说明是根节点，需替换为leftPage，因为只允许根节点的条目数为0，其他都为非0，否则应该被合并了已经
			// 说明此前只有一个条目，指向左右两个节点，右节点已经被删掉了，只能让左节点成为根
			BTreePageId rootPtrId = parent.getParentId();
			if(rootPtrId.pgcateg() != BTreePageId.ROOT_PTR) {
				throw new DbException("attempting to delete a non-root node");
			}
			// 获取根指针页面，更新根节点为leftPage
			BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, rootPtrId, Permissions.READ_WRITE);
			leftPage.setParentId(rootPtrId);
			rootPtr.setRootId(leftPage.getId());
			System.out.printf("[TID" + tid.getId() + "] 父节点ID-%d 为根节点，创建新的根节点并修改原根节点为叶子节点\n", parent.getId().getPageNumber());
			// release the parent page for reuse
			// 释放原父节点页面（标记为空闲，可重用）
			setEmptyPage(tid, dirtypages, parent.getId().getPageNumber());
		}
		// 填充率过低，意味非根，直接正常处理
		else if(parent.getNumEmptySlots() > maxEmptySlots) {
			System.out.printf("[TID" + tid.getId() + "] [deleteParentEntry]节点ID-%d 删除后触发融合操作\n", parent.getId().getPageNumber());
			// System.out.println("触发非页节点融合操作");
			handleMinOccupancyPage(tid, dirtypages, parent);
		}else {
			System.out.printf("[TID" + tid.getId() + "] [deleteParentEntry]父节点ID-%d 删除节点后节点数量变为%d, 不触发融合操作\n", parent.getId().getPageNumber(), parent.getNumEntries());
		}
	}

	/**
	 * Delete a tuple from this BTreeFile. 
	 * May cause pages to merge or redistribute entries/tuples if the pages 
	 * become less than half full.
	 * 
	 * @param tid - the transaction id
	 * @param t - the tuple to delete
	 * @return a list of all pages that were dirtied by this operation. Could include
	 * many pages since parent pointers will need to be updated when an internal node merges.
	 * @see #handleMinOccupancyPage(TransactionId, Map, BTreePage)
	 */
	public synchronized List<Page> deleteTuple(TransactionId tid, Tuple t)
			throws DbException, IOException, TransactionAbortedException {
		Map<PageId, Page> dirtypages = new HashMap<>();

		BTreePageId pageId = new BTreePageId(tableid, t.getRecordId().getPageId().getPageNumber(),
				BTreePageId.LEAF);
		// 从缓存或缓冲池获取该叶子页面，申请读写权限（因需要删除元组）
		BTreeLeafPage page = (BTreeLeafPage) getPage(tid, dirtypages, pageId, Permissions.READ_WRITE);
		// 执行元组删除操作
		page.deleteTuple(t);
		// System.out.println ("删除后叶节点条目数为:" + page.getNumTuples());
		System.out.println("[TID" + tid.getId() + "] 删除后叶节点ID-" + page.getId().getPageNumber() + " 条目数为:" + page.getNumTuples());

		// if the page is below minimum occupancy, get some tuples from its siblings
		// or merge with one of the siblings
		int maxEmptySlots = page.getMaxTuples() - page.getMaxTuples()/2; // ceiling
		// // 若页面空槽位超过阈值（填充率低于一半），触发平衡操作
		if(page.getNumEmptySlots() > maxEmptySlots) {
			// 输出节点信息和空槽位情况
			System.out.println("[TID" + tid.getId() + "] 节点 " + page.getId().getPageNumber() + " 空槽位数量(" + page.getNumEmptySlots() + ")超过阈值(" + maxEmptySlots + ")，触发填充率调整");
			handleMinOccupancyPage(tid, dirtypages, page);
		}

        return new ArrayList<>(dirtypages.values());
	}

	/**
	 * Get a read lock on the root pointer page. Create the root pointer page and root page
	 * if necessary.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages 
	 * @return the root pointer page
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	BTreeRootPtrPage getRootPtrPage(TransactionId tid, Map<PageId, Page> dirtypages) throws DbException, IOException, TransactionAbortedException {
		synchronized(this) {
			// 如果存储 B 树的文件是空的（首次创建 B 树），需要初始化两个关键页面：
			// 根指针页面（BTreeRootPtrPage）：通过 createEmptyPageData() 生成空数据，用于后续存储根节点 ID。
			// 初始根页面：默认创建一个空的叶子节点（BTreeLeafPage），作为 B 树的第一个根节点（B 树初始时根节点为叶子节点）。
			// 将两个空页面的数据写入文件，完成 B 树的初始化
			if(f.length() == 0) {
				// create the root pointer page and the root page
				BufferedOutputStream bw = new BufferedOutputStream(
						new FileOutputStream(f, true));
				byte[] emptyRootPtrData = BTreeRootPtrPage.createEmptyPageData();
				byte[] emptyLeafData = BTreeLeafPage.createEmptyPageData();
				bw.write(emptyRootPtrData);
				bw.write(emptyLeafData);
				bw.close();
			}
		}
		// 如果已经创建过了就不用创建，直接读取
		// get a read lock on the root pointer page
		// 调用 getPage 方法，以只读权限（Permissions.READ_ONLY） 获取该页面
		return (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid), Permissions.READ_ONLY);
	}

	/**
	 * Get the page number of the first empty page in this BTreeFile.
	 * Creates a new page if none of the existing pages are empty.
	 * 获取当前 B 树文件（BTreeFile）中第一个空页面的页码（页面编号）。如果所有现有页面都已被使用（无空页面），则创建一个新页面
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @return the page number of the first empty page
	 * 获取空页面用于分配
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public int getEmptyPageNo(TransactionId tid, Map<PageId, Page> dirtypages)
			throws DbException, IOException, TransactionAbortedException {
		// get a read lock on the root pointer page and use it to locate the first header page
		//  定位头页面（BTreeHeaderPage）的入口
		BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
		BTreePageId headerId = rootPtr.getHeaderId();
		int emptyPageNo = 0;

		//  遍历头页面链，寻找有空闲槽位的头页面
		if(headerId != null) {
			// 先以只读权限获取第一个头页面
			BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
			int headerPageCount = 0;
			// try to find a header page with an empty slot
			// // 循环遍历头页面链：如果当前头页面没有空槽位，就找下一个头页面
			while(headerPage != null && headerPage.getEmptySlot() == -1) {
				headerId = headerPage.getNextPageId();
				if(headerId != null) {
					headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
					headerPageCount++;
				}
				else {
					headerPage = null; // 没有更多头页面，退出循环
				}
			}

			// if headerPage is not null, it must have an empty slot
			// 如果找到有空闲槽位的头页面
			if(headerPage != null) {
				// 升级权限为读写（需修改槽位状态）
				headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_WRITE);
				int emptySlot = headerPage.getEmptySlot(); // 获取空闲槽位索引

				headerPage.markSlotUsed(emptySlot, true); // 将该槽位标记为“已使用”（避免重复分配）
				// 计算空页面页码：头页面数量×每个头页面的槽位数 + 槽位索引
				emptyPageNo = headerPageCount * BTreeHeaderPage.getNumSlots() + emptySlot;
			}
		}

		// at this point if headerId is null, either there are no header pages 
		// or there are no free slots
		// 若headerId为null（无可用头页面或所有头页面无空闲槽位），则创建新页面
		if(headerId == null) {		
			synchronized(this) { // 同步块：确保多线程下创建新页面的原子性
				// create the new page
				// 向文件末尾写入一个空页面的数据（默认创建内部节点的空数据）
				BufferedOutputStream bw = new BufferedOutputStream(
						new FileOutputStream(f, true));
				byte[] emptyData = BTreeInternalPage.createEmptyPageData();
				bw.write(emptyData);
				bw.close();
				// 新页面的页码为当前总页面数（因文件末尾新增一页，页码递增）
				emptyPageNo = numPages();
			}
		}

		return emptyPageNo; 
	}
	
	/**
	 * Method to encapsulate the process of creating a new page.  It reuses old pages if possible,
	 * and creates a new page if none are available.  It wipes the page on disk and in the cache and 
	 * returns a clean copy locked with read-write permission
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param pgcateg - the BTreePageId category of the new page.  Either LEAF, INTERNAL, or HEADER
	 * @return the new empty page
	 * @see #getEmptyPageNo(TransactionId, Map)
	 * @see #setEmptyPage(TransactionId, Map, int)
	 * 封装创建新页面的过程。优先重用旧的空闲页面，若没有可用空闲页面则创建新页面。
	 * 创建 / 重用时会清除磁盘和缓存中该页面的旧数据，并返回一个带有读写权限的干净页面副本。
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	private Page getEmptyPage(TransactionId tid, Map<PageId, Page> dirtypages, int pgcateg)
			throws DbException, IOException, TransactionAbortedException {
		// create the new page
		// 调用getEmptyPageNo方法获取一个空闲页面的页码
		int emptyPageNo = getEmptyPageNo(tid, dirtypages);
		// 创建页面id
		BTreePageId newPageId = new BTreePageId(tableid, emptyPageNo, pgcateg);
		
		// write empty page to disk
		// 打开文件以读写模式操作
		RandomAccessFile rf = new RandomAccessFile(f, "rw");
		// 计算页面在文件中的偏移量：根指针页面大小 + (页码-1)×标准页面大小
		rf.seek(BTreeRootPtrPage.getPageSize() + (long) (emptyPageNo - 1) * BufferPool.getPageSize());
		// 写入空页面数据（覆盖旧数据）
		rf.write(BTreePage.createEmptyPageData());
		rf.close();
		
		// make sure the page is not in the buffer pool	or in the local cache
		// 从缓冲池丢弃该页面的旧版本
		Database.getBufferPool().discardPage(newPageId);
		// 从本地脏页缓存中移除该页面（若存在）
		dirtypages.remove(newPageId);
		// 获取带有读写权限的新页面
		return getPage(tid, dirtypages, newPageId, Permissions.READ_WRITE);
	}

	/**
	 * Mark a page in this BTreeFile as empty. Find the corresponding header page 
	 * (create it if needed), and mark the corresponding slot in the header page as empty.
	 * 
	 * @param tid - the transaction id
	 * @param dirtypages - the list of dirty pages which should be updated with all new dirty pages
	 * @param emptyPageNo - the page number of the empty page
	 * @see #getEmptyPage(TransactionId, Map, int)
	 * 将 B 树文件（BTreeFile）中的指定页面标记为 “空页面”（空闲页面）。
	 * 具体操作是找到该页面对应的头页面（BTreeHeaderPage，若不存在则创建），
	 * 并在头页面中将对应槽位标记为空闲，以便后续重用。
	 * 负责将不再使用的页面（如合并后废弃的页面）标记为空闲
	 * @throws DbException
	 * @throws IOException
	 * @throws TransactionAbortedException
	 */
	public void setEmptyPage(TransactionId tid, Map<PageId, Page> dirtypages, int emptyPageNo)
			throws DbException, IOException, TransactionAbortedException {

		// if this is the last page in the file (and not the only page), just 
		// truncate the file
		// @TODO: Commented out because we should probably do this somewhere else in case the transaction aborts....
//		synchronized(this) {
//			if(emptyPageNo == numPages()) {
//				if(emptyPageNo <= 1) {
//					// if this is the only page in the file, just return.
//					// It just means we have an empty root page
//					return;
//				}
//				long newSize = f.length() - BufferPool.getPageSize();
//				FileOutputStream fos = new FileOutputStream(f, true);
//				FileChannel fc = fos.getChannel();
//				fc.truncate(newSize);
//				fc.close();
//				fos.close();
//				return;
//			}
//		}

		// otherwise, get a read lock on the root pointer page and use it to locate 
		// the first header page
		// 获取根指针页面（读锁），定位头页面ID
		System.out.printf("[TID" + tid.getId() + "] [setEmptyPage] 准备清空节点%d\n", emptyPageNo);
		BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
		BTreePageId headerId = rootPtr.getHeaderId();
		BTreePageId prevId = null;// 记录上一个头页面的ID
		int headerPageCount = 0;// 记录已遍历的头页面数量

		// if there are no header pages, create the first header page and update
		// the header pointer in the BTreeRootPtrPage
		// 若没有头页面（headerId为null），则创建第一个头页面并更新根指针
		if(headerId == null) {
			// 升级根指针页面权限为读写（需修改头页面ID）
			rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid), Permissions.READ_WRITE);
			// 创建一个空的头页面（类型为HEADER）
			BTreeHeaderPage headerPage = (BTreeHeaderPage) getEmptyPage(tid, dirtypages, BTreePageId.HEADER);
			headerId = headerPage.getId();
			headerPage.init();// 初始化头页面（如设置槽位默认状态）
			rootPtr.setHeaderId(headerId);
		}

		// iterate through all the existing header pages to find the one containing the slot
		// corresponding to emptyPageNo
		// 遍历现有头页面链，找到负责管理emptyPageNo的头页面
		while(headerId != null && (headerPageCount + 1) * BTreeHeaderPage.getNumSlots() < emptyPageNo) {
			BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
			prevId = headerId; // 记录当前头页面ID（作为下一个的前序）
			headerId = headerPage.getNextPageId();// 获取下一个头页面ID
			headerPageCount++;// 累加头页面数量
		}

		// at this point headerId should either be null or set with 
		// the headerPage containing the slot corresponding to emptyPageNo.
		// Add header pages until we have one with a slot corresponding to emptyPageNo
		// 若现有头页面链的槽位总数仍小于emptyPageNo，继续创建新头页面扩展链
		while((headerPageCount + 1) * BTreeHeaderPage.getNumSlots() < emptyPageNo) {
			// 获取上一个头页面（读写权限）
			BTreeHeaderPage prevPage = (BTreeHeaderPage) getPage(tid, dirtypages, prevId, Permissions.READ_WRITE);
			// 创建新的头页面
			BTreeHeaderPage headerPage = (BTreeHeaderPage) getEmptyPage(tid, dirtypages, BTreePageId.HEADER);
			headerId = headerPage.getId();
			headerPage.init();
			headerPage.setPrevPageId(prevId); // 新头页面记录前序头页面ID
			prevPage.setNextPageId(headerId); // 前序头页面记录新头页面ID（形成链表）

			headerPageCount++; // 头页面数量+1
			prevId = headerId; // 更新前序ID为新头页面
		}

		// now headerId should be set with the headerPage containing the slot corresponding to 
		// emptyPageNo
		// 获取目标头页面（读写权限）
		BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_WRITE);
		// 计算目标页码在当前头页面中的槽位索引：页码 - 已遍历头页面的总槽位数
		int emptySlot = emptyPageNo - headerPageCount * BTreeHeaderPage.getNumSlots();
		// 将该槽位标记为“未使用”（即空闲）
		headerPage.markSlotUsed(emptySlot, false);
		System.out.printf("[TID" + tid.getId() + "] [setEmptyPage] 节点%d清空完毕\n", emptyPageNo);
	}

	/**
	 * get the specified tuples from the file based on its IndexPredicate value on
	 * behalf of the specified transaction. This method will acquire a read lock on
	 * the affected pages of the file, and may block until the lock can be
	 * acquired.
	 * 据索引条件从文件中查询符合条件的元组（tuples），并返回一个迭代器用于遍历这些结果
	 * @param tid - the transaction id
	 * @param ipred - the index predicate value to filter on
	 * @return an iterator for the filtered tuples
	 */
	public DbFileIterator indexIterator(TransactionId tid, IndexPredicate ipred) {
		return new BTreeSearchIterator(this, tid, ipred);
	}

	/**
	 * Get an iterator for all tuples in this B+ tree file in sorted order. This method 
	 * will acquire a read lock on the affected pages of the file, and may block until 
	 * the lock can be acquired.
	 * 获取该 B + 树文件中所有元组（tuples）的迭代器，并且这些元组是以排序顺序返回的。
	 * @param tid - the transaction id
	 * @return an iterator for all the tuples in this file
	 */
	public DbFileIterator iterator(TransactionId tid) {
		return new BTreeFileIterator(this, tid);
	}

}

/**
 * Helper class that implements the Java Iterator for tuples on a BTreeFile
 * 通过 B + 树叶子节点形成的有序链表结构，按索引键的顺序逐个返回所有元组，支持遍历、重置和关闭操作
 */
class BTreeFileIterator extends AbstractDbFileIterator {

	Iterator<Tuple> it = null;
	BTreeLeafPage curp = null;

	final TransactionId tid;
	final BTreeFile f;

	/**
	 * Constructor for this iterator
	 * @param f - the BTreeFile containing the tuples
	 * @param tid - the transaction id
	 */
	public BTreeFileIterator(BTreeFile f, TransactionId tid) {
		this.f = f;
		this.tid = tid;
	}

	/**
	 * Open this iterator by getting an iterator on the first leaf page
	 * 初始化迭代器，定位到 B + 树的第一个叶子页，准备开始遍历。
	 */
	public void open() throws DbException, TransactionAbortedException {
		BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) Database.getBufferPool().getPage(
				tid, BTreeRootPtrPage.getId(f.getId()), Permissions.READ_ONLY);
		BTreePageId root = rootPtr.getRootId();
		curp = f.findLeafPage(tid, root, null);
		it = curp.iterator();
	}

	/**
	 * Read the next tuple either from the current page if it has more tuples or
	 * from the next page by following the right sibling pointer.
	 * 按顺序返回下一个元组，若当前页遍历完毕则自动切换到下一个叶子页，通过rightSiblingId（右兄弟指针）获取下一个叶子页
	 * @return the next tuple, or null if none exists
	 */
	@Override
	protected Tuple readNext() throws TransactionAbortedException, DbException {
		if (it != null && !it.hasNext())
			it = null;

		while (it == null && curp != null) {
			BTreePageId nextp = curp.getRightSiblingId();
			if(nextp == null) {
				curp = null;
			}
			else {
				curp = (BTreeLeafPage) Database.getBufferPool().getPage(tid,
						nextp, Permissions.READ_ONLY);
				it = curp.iterator();
				if (!it.hasNext())
					it = null;
			}
		}

		if (it == null)
			return null;
		return it.next();
	}

	/**
	 * rewind this iterator back to the beginning of the tuples
	 */
	public void rewind() throws DbException, TransactionAbortedException {
		close();
		open();
	}

	/**
	 * close the iterator
	 */
	public void close() {
		super.close();
		it = null;
		curp = null;
	}
}

/**
 * Helper class that implements the DbFileIterator for search tuples on a
 * B+ Tree File
 * 它是 B + 树文件（BTreeFile）中根据索引条件（IndexPredicate）筛选元组的迭代器
 */
class BTreeSearchIterator extends AbstractDbFileIterator {

	Iterator<Tuple> it = null;
	BTreeLeafPage curp = null;

	final TransactionId tid;
	final BTreeFile f;
	final IndexPredicate ipred;

	/**
	 * Constructor for this iterator
	 * @param f - the BTreeFile containing the tuples
	 * @param tid - the transaction id
	 * @param ipred - the predicate to filter on
	 */
	public BTreeSearchIterator(BTreeFile f, TransactionId tid, IndexPredicate ipred) {
		this.f = f;
		this.tid = tid;
		this.ipred = ipred;
	}

	/**
	 * Open this iterator by getting an iterator on the first leaf page applicable
	 * for the given predicate operation
	 * 若操作是EQUALS（等于）、GREATER_THAN（大于）、GREATER_THAN_OR_EQ（大于等于）：
	 * 调用findLeafPage并传入ipred.getField()（目标字段值），直接定位到第一个可能符合条件的叶子页（利用 B + 树的索引快速定位，避免从第一个页开始遍历）。
	 * 若操作是LESS_THAN（小于）、LESS_THAN_OR_EQ（小于等于）：从 B + 树的第一个叶子页开始遍历（因为 “小于” 条件的元组集中在左侧）
	 */
	public void open() throws DbException, TransactionAbortedException {
		BTreeRootPtrPage rootPtr = null;
		rootPtr = (BTreeRootPtrPage) Database.getBufferPool().getPage(
					tid, BTreeRootPtrPage.getId(f.getId()), Permissions.READ_ONLY);
		BTreePageId root = rootPtr.getRootId();
		if(ipred.getOp() == Op.EQUALS || ipred.getOp() == Op.GREATER_THAN 
				|| ipred.getOp() == Op.GREATER_THAN_OR_EQ) {
			curp = f.findLeafPage(tid, root, ipred.getField());
		}
		else {
			curp = f.findLeafPage(tid, root, null);
		}
		if (curp!=null) {
			System.out.println("[TID" + tid.getId() + "] 初始化过滤器成功");
		}else {
			System.out.println("[TID" + tid.getId() + "] 初始化过滤器失败");
		}
		it = curp.iterator();
	}

	/**
	 * Read the next tuple either from the current page if it has more tuples matching
	 * the predicate or from the next page by following the right sibling pointer.
	 * 用t.getField(f.keyField()).compare(ipred.getOp(), ipred.getField())判断元组是否符合条件：
	 * 若符合，直接返回该元组。
	 * 若不符合，根据操作类型（Op）判断是否需要继续遍历：
	 * 对于LESS_THAN/LESS_THAN_OR_EQ（小于 / 小于等于）：B + 树叶子页是有序的，一旦遇到不符合的元组（如键值已大于目标值），说明后续元组都不符合，直接返回null。
	 * 对于EQUALS（等于）：若当前元组键值已大于目标值（compare(Op.GREATER_THAN, ...)为真），说明后续元组都大于目标值，不可能符合 “等于” 条件，返回null。
	 * 若当前页已遍历完（it无更多元组），通过叶子页的rightSiblingId（右兄弟指针）获取下一个叶子页
	 * @return the next tuple matching the predicate, or null if none exists
	 */
	@Override
	protected Tuple readNext() throws TransactionAbortedException, DbException,
	NoSuchElementException {
		while (it != null) {

			while (it.hasNext()) {
				Tuple t = it.next();
				if (t.getField(f.keyField()).compare(ipred.getOp(), ipred.getField())) {
					return t;
				}
				else if(ipred.getOp() == Op.LESS_THAN || ipred.getOp() == Op.LESS_THAN_OR_EQ) {
					// if the predicate was not satisfied and the operation is less than, we have
					// hit the end
					return null;
				}
				else if(ipred.getOp() == Op.EQUALS && 
						t.getField(f.keyField()).compare(Op.GREATER_THAN, ipred.getField())) {
					// if the tuple is now greater than the field passed in and the operation
					// is equals, we have reached the end
					return null;
				}
			}

			BTreePageId nextp = curp.getRightSiblingId();
			// if there are no more pages to the right, end the iteration
			if(nextp == null) {
				return null;
			}
			else {
				curp = (BTreeLeafPage) Database.getBufferPool().getPage(tid,
						nextp, Permissions.READ_ONLY);
				it = curp.iterator();
			}
		}

		return null;
	}

	/**
	 * rewind this iterator back to the beginning of the tuples
	 */
	public void rewind() throws DbException, TransactionAbortedException {
		close();
		open();
	}

	/**
	 * close the iterator
	 */
	public void close() {
		super.close();
		it = null;
	}
}
